package com.ndp.fb.mq.consumer.message;

import com.alibaba.dubbo.config.annotation.Reference;
import com.ndp.fb.constant.ConfigConst;
import com.ndp.fb.enums.message.ChannelType;
import com.ndp.fb.enums.message.EventNotificationType;
import com.ndp.fb.enums.message.MessageStatus;
import com.ndp.fb.model.fbagent.param.ApiCallBackTarget;
import com.ndp.fb.mq.api.producer.RocketMqProducerService;
import com.ndp.fb.mq.api.service.messageservice.EventNotificationMessageService;
import com.ndp.fb.mq.consumer.AbstractPullConsumer;
import com.ndp.fb.rdb.api.UserRdbServices;
import com.ndp.fb.rdb.model.User;
import com.ndp.fb.rdb.model.message.EventNotificationMessage;
import com.ndp.fb.util.ThreadPool;
import com.ndp.fb.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * Created by elvisxiong on 7/24/15.
 */
@Component
public class MessageConsumerImpl extends AbstractPullConsumer {

    private static Logger logger = LoggerFactory.getLogger(MessageConsumerImpl.class);

    @Autowired
    EventNotificationMessageService eventNotificationMessageService;

    @Autowired
    RocketMqProducerService rocketMqProducerService;

    @Reference
    UserRdbServices userRdbServices;

    public static void main(String[] args) {
        //
        ApiCallBackTarget.MessageCenter target = (ApiCallBackTarget.MessageCenter) ApiCallBackTarget.getTargetType("message_center_topic$$$$$$EventNotificationMessage");
        System.out.println(target);
    }

    @Override
    public String getTopicName() {
        return ConfigConst.MQ_SERVICES_MESSAGE_CENTER_TOPIC;
    }

    @Override
    protected void consume(String tag, String keys, Object message, Map<String, String> equipment) {
        logger.info("MessageConsumerImpl tag >>>> " + tag + " keys >>>" + keys + " message >>> " + message);
        ApiCallBackTarget.MessageCenter target = (ApiCallBackTarget.MessageCenter) ApiCallBackTarget.getTargetType(tag);
        switch (target) {
            case EventNotificationMessage:
                final String eventType = equipment.get("eventType");
                final String isNeedFunctionJump = equipment.get("isNeedFunctionJump");
                final String url = equipment.get("url");
                final String sender = equipment.get("sender");
                final String receiver = equipment.get("receiver");
                final String body = (String) message;
                final String channelTypes = equipment.get("channelTypes");
                final String eventKey = equipment.get("eventKey");
                final String eventValues = equipment.get("eventValues");
                final String[] userIdArray = receiverToUserIdArray(receiver);
                Set<String> setUserIds = new TreeSet<String>();
                for (String userId : userIdArray) {
                    setUserIds.add(userId);
                }

                if (setUserIds != null) {
                    int i = 0;
                    for (String str : setUserIds) {
                        final int j = i;
                        final String userId = str;
                        ThreadPool.execute(() -> {
                            Long startTime = System.currentTimeMillis();
                            eventNotificationMessageSend(Integer.parseInt(eventType), isNeedFunctionJump, url, sender, userId, body,
                                    channelTypes, j,eventKey,eventValues);
                            Long endTime = System.currentTimeMillis();
                            logger.info("保存并发送到walle websocket消息队列共消耗：" + (endTime - startTime) / 1000 + "s");
                        });
                        i++;
                    }
                    
                }
                break;
            case EventNotificationMonitorMessage:
            	EventNotificationMessage eventNotificationMessage = (EventNotificationMessage)message;
            	List<String> channelTypeList = CollectionUtils.arrayToList(equipment.get("chanel").split(","));
                eventNotificationMessageService.sendMonitorMessage(eventNotificationMessage, channelTypeList);
                break;
            default:
                break;
        }
        ;
    }

    private void eventNotificationMessageSend(int eventType, String isNeedFunctionJump, String url, String sender, String userId, String body, String channelTypes,
                                              int i,String eventKey,String eventValues) {
        EventNotificationMessage eventNotificationMessage = new EventNotificationMessage();
        EventNotificationType eventNotificationType = EventNotificationType.getEventNotificationType(eventType);
        eventNotificationMessage.setEventNo(UUID.get());
        eventNotificationMessage.setEventType(eventNotificationType.getType());
        if (isNeedFunctionJump != null && isNeedFunctionJump.equals("true")) {
            eventNotificationMessage.setIsNeedFunctionJump(true);
        } else {
            eventNotificationMessage.setUrl(url);
        }
        eventNotificationMessage.setUrl(url);
        eventNotificationMessage.setSender(sender);
        eventNotificationMessage.setReceiver(userId);
        eventNotificationMessage.setIsRead(0);
        eventNotificationMessage.setBody(body);
        eventNotificationMessage.setSendTime(System.currentTimeMillis());
        eventNotificationMessage.setStatus(MessageStatus.NORMAL.getStatus()); //todo verify to elvis
        eventNotificationMessage.setEventKey(eventKey);
        eventNotificationMessage.setEventValues(eventValues);

        List<ChannelType> channelTypeList = new ArrayList<ChannelType>();
        if (channelTypes != null && !channelTypes.equals("")) {

            String[] channelArray = channelTypes.split(",");
            for (int j = 0; j < channelArray.length; j++) {
                channelTypeList.add(ChannelType.getChannelType(Integer.parseInt(channelArray[j])));
            }
        } else {
            channelTypeList.add(ChannelType.WEBSITE);
        }
        ;
        eventNotificationMessageService.sendMessage(eventNotificationMessage, channelTypeList);
    }

    private String[] receiverToUserIdArray(String receiver) {
        StringBuffer userIds = new StringBuffer();
        if (receiver == null || receiver.equals("")) {
            List<User> users = userRdbServices.getAllUsers();
            if (users != null) {
                for (int i = 0; i < users.size(); i++) {
                    userIds.append(users.get(i).getTicket());
                    if (i < users.size() - 1) {
                        userIds.append(",");
                    }
                }
            }
        } else {
            userIds.append(receiver);
        }
        return userIds.toString().split(",");
    }
}
